package com.stp.kafka.controller;

import com.stp.kafka.callback.KafkaTestListenableFutureCallback;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.util.concurrent.SuccessCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class KafkaController {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * @KafkaListener注解所标注的方法并不会在IOC容器中被注册为Bean，
     * 而是会被注册在KafkaListenerEndpointRegistry中，
     * 而KafkaListenerEndpointRegistry在SpringIOC中已经被注册为Bean
     **/
    @Autowired
    private KafkaListenerEndpointRegistry registry;


    @GetMapping("/kafka/normal/{message}")
    public void sendNormalMessage(@PathVariable("message") String message) {
        kafkaTemplate.send("sb_topic", message);//.addCallback(new KafkaTestListenableFutureCallback());
    }

    /**
     * 回调的第一种写法
     * @param message
     */
    @GetMapping("/kafka/callbackOne/{message}")
    public void sendCallbackOneMessage(@PathVariable("message") String message) {
        kafkaTemplate.send("sb_topic", message)
                .addCallback(new SuccessCallback<SendResult<String, Object>>() {
            //成功的回调
            @Override
            public void onSuccess(SendResult<String, Object> success) {
                // 消息发送到的topic
                String topic = success.getRecordMetadata().topic();
                // 消息发送到的分区
                int partition = success.getRecordMetadata().partition();
                // 消息在分区内的offset
                long offset = success.getRecordMetadata().offset();
                System.out.println("发送消息成功1:" + topic + "-" + partition + "-" + offset);
            }
        }, new FailureCallback() {
            //失败的回调
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println("发送消息失败1:" + throwable.getMessage());
            }
        });
    }

    /**
     * 回调的第二种写法
     * @param message
     */
    @GetMapping("/kafka/callbackTwo/{message}")
    public void sendCallbackTwoMessage(@PathVariable("message") String message) {
        kafkaTemplate.send("sb_topic", message).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println("发送消息失败2："+throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                System.out.println("发送消息成功2：" + result.getRecordMetadata().topic() + "-"
                        + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
            }
        });
    }


    /**
     * 需要在yml spring.kafka.producer.transaction-id-prefix: tx_ #事务id前缀
     * @param message
     */
    @GetMapping("/kafka/transaction/{message}")
    public void sendTransactionMessage(@PathVariable("message") String message) {
        //声明事务：后面报错消息不会发出去
        kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<String, Object, Object>() {

            @Override
            public Object doInOperations(KafkaOperations<String, Object> operations) {
                operations.send("sb_topic", message + " test executeInTransaction");
                throw new RuntimeException("fail");
            }
        });
        // //不声明事务：后面报错但前面消息已经发送成功了
        // kafkaTemplate.send("sb_topic", message + " test executeInNoTransaction");
        // throw new RuntimeException("fail");
    }

    /**
     * 开启某一个监听器
     * @param listenerId 监听器ID
     */
    @GetMapping("/kafka/listener/start/{listenerId}")
    public void startListener(@PathVariable("listenerId")String listenerId){
        MessageListenerContainer consumer = registry.getListenerContainer(listenerId);
        if (!consumer.isRunning()) {
            consumer.start();
        }
    }

    /**
     * 暂停某一个监听器
     * @param listenerId 监听器ID
     */
    @GetMapping("/kafka/listener/pause/{listenerId}")
    public void pauseListener(@PathVariable("listenerId")String listenerId){
        MessageListenerContainer consumer = registry.getListenerContainer(listenerId);
        if (consumer.isRunning()) {
            consumer.pause();
        }
    }

}
